use crate::cache::get_pg_pool;
use crate::keys::{BoundTypeKeyTrait, IndexTypeKeyTrait, UniqueTypeKeyTrait, VddKey};
use crate::record_db::RecordDbTrait;
use crate::sqlx::get_sql;
use futures::TryStreamExt;
use serde_json::json;

use sqlx::{Connection, Executor, PgPool, Postgres, Row};
use std::collections::HashMap;
use std::sync::Arc;
use visit_dd_core::field::{VddField, VddFieldRef};
use visit_dd_core::i18n_text::E_WRONG_KEY_TYPE;
use visit_dd_core::model::{DataStore, DataStoreEngine, VddModel, VddSchema};
use visit_dd_core::proj::VddUser;
use visit_dd_core::query::{
    QueryDataPage, QueryFilter, QueryPlan, ShareQueryParams, WebQueryParams,
};
use visit_dd_core::record::{VddDisplay, VddRecord};
use visit_dd_core::{i18n, to_sha256_hash, Ares, Astr, FieldId, SqlAttribute};

use super::SqlxDefineTrait;

impl SqlxDefineTrait for PgPool {
    async fn update_entity(&self, id: u64, body: &str) -> Ares {
        let r = sqlx::query(&get_sql("update_entity", DataStoreEngine::Postgres))
            .bind(body)
            .bind(id as i64)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn load_entity(&self, id: u64) -> Result<Astr, Astr> {
        let r = sqlx::query(&get_sql("load_entity", DataStoreEngine::Postgres))
            .bind(id as i64)
            .fetch_one(self)
            .await;
        match r {
            Ok(r2) => Ok(r2.get::<&str, _>("entity_content").into()),
            Err(e) => Err(e.to_string().into()),
        }
    }

    async fn load_entities(&self, vk: VddKey, deleted: bool) -> Result<Vec<Astr>, Astr> {
        if !vk.entity_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let rows = sqlx::query(&get_sql("load_entities", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(&vk) as i16)
            .bind(if deleted { 1 } else { 0 })
            .fetch_all(self)
            .await
            .unwrap();
        let mut r_vec = Vec::<Astr>::new();
        rows.iter().for_each(|row| {
            r_vec.push(row.try_get::<&str, &str>("entity_content").unwrap().into());
        });
        Ok(r_vec)
    }

    async fn load_entities_form_index_query(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        deleted: bool,
    ) -> Result<Vec<Astr>, Astr> {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = if vk.get_index_key_as_u64() > 0 {
            get_sql("load_entities_by_index", DataStoreEngine::Postgres)
        } else {
            get_sql("load_entities_by_index_no_arg", DataStoreEngine::Postgres)
        };
        let rows = if vk.get_index_key_as_u64() > 0 {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_index_key_as_u64() as i64)
                .bind(if deleted { 1 } else { 0 })
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(if deleted { 1 } else { 0 })
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r_vec = Vec::<Astr>::new();
        rows.iter().for_each(|row| {
            r_vec.push(row.try_get::<&str, _>("entity_content").unwrap().into());
        });
        Ok(r_vec)
    }

    async fn count_entities(&self, vk: VddKey) -> Result<usize, Astr> {
        if !vk.entity_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let row = sqlx::query(&get_sql("count_entities", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(&vk) as i16)
            .fetch_one(self)
            .await
            .unwrap();
        let r = row.get::<i64, &str>("count");
        Ok(r as usize)
    }

    async fn delete_entity(&self, id: u64) -> Ares {
        let r = sqlx::query(&get_sql("delete_entity", DataStoreEngine::Postgres))
            .bind(id as i64)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_entities_from_index_query(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Ares {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let r = sqlx::query(&get_sql(
            "delete_entities_from_index_query",
            DataStoreEngine::Postgres,
        ))
        .bind(Into::<u16>::into(vk.as_ref()) as i16)
        .bind(vk.get_index_key_as_u64() as i64)
        .execute(self)
        .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn add_to_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        index_item: u64,
    ) -> Ares {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = get_sql("insert_index", DataStoreEngine::Postgres);
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(index_item as i64)
            .bind(vk.get_index_key_as_u64() as i64)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        index_item: u64,
    ) -> Ares {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = get_sql("delete_index", DataStoreEngine::Postgres);
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(index_item as i64)
            .bind(vk.get_index_key_as_u64() as i16)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_indexes(&self, vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>) -> Ares {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let r = sqlx::query(&get_sql("delete_indexes", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_index_key_as_u64() as i16)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn read_index(&self, vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>) -> Vec<i64> {
        if !vk.index_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let sql_name = if vk.get_index_key_as_u64() > 0 {
            "read_indexes"
        } else {
            "read_indexes_no_arg"
        };
        let rows = if vk.get_index_key_as_u64() > 0 {
            sqlx::query(&get_sql(sql_name, DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_index_key_as_u64() as i64)
                .fetch_all(self)
                .await
        } else {
            sqlx::query(&get_sql(sql_name, DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .fetch_all(self)
                .await
        };
        let mut r_vec = Vec::<i64>::new();
        rows.unwrap().iter().for_each(|row| {
            let item = row.get::<i64, &str>("index_item");
            r_vec.push(item);
        });
        r_vec
    }

    async fn count_in_index(
        &self,
        vk: impl IndexTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<usize, Astr> {
        if !vk.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let row = sqlx::query(&get_sql("count_in_index", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_index_key_as_u64() as i64)
            .fetch_one(self)
            .await
            .unwrap();
        let r = row.get::<i64, &str>("count");
        Ok(r as usize)
    }

    async fn get_unique(
        &self,
        vk: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Result<u64, Astr> {
        if !vk.unique_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let row = sqlx::query(&get_sql("get_unique", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_unique_hash().unwrap().as_ref())
            .fetch_one(self)
            .await;
        match row {
            Ok(row2) => Ok(row2.get::<i64, _>("link_id") as u64),
            Err(e) => Err(e.to_string().into()),
        }
    }

    async fn delete_unique_by_value(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        link_id: u64,
    ) -> Ares {
        if !vk_u.unique_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = get_sql("delete_unique", DataStoreEngine::Postgres);
        let r = sqlx::query(&sql)
            .bind(Into::<u16>::into(vk_u.as_ref()) as i16)
            .bind(link_id as i64)
            .execute(self)
            .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn delete_uniques_from_index_query(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        vk_i: VddKey,
    ) -> Ares {
        if !vk_u.unique_type() || !vk_i.index_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let r = sqlx::query(&get_sql(
            "delete_uniques_from_index_query",
            DataStoreEngine::Postgres,
        ))
        .bind(Into::<u16>::into(vk_i.as_ref()) as i16)
        .bind(vk_i.get_index_key_as_u64() as i64)
        .bind(Into::<u16>::into(vk_u.as_ref()) as i16)
        .execute(self)
        .await;
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn select_uniques(
        &self,
        vk: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> HashMap<String, u64> {
        if !vk.unique_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let hash = vk.get_unique_hash();
        let sql = if hash.is_ok() {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = $1 AND hash = $2"
        } else {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = $1"
        };
        let rows = if hash.is_ok() {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(hash.unwrap().to_string())
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                row.get::<String, &str>("hash"),
                row.get::<i64, &str>("link_id") as u64,
            );
        });
        r
    }

    async fn select_uniques_by_value(
        &self,
        vk: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        link_id: Option<u64>,
    ) -> HashMap<String, u64> {
        if !vk.unique_type() {
            panic!("{}", i18n!(E_WRONG_KEY_TYPE));
        }
        let sql = if link_id.is_some() {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = $1 AND link_id = $2"
        } else {
            "SELECT hash,link_id FROM t_unique WHERE unique_type = $1"
        };
        let rows = if link_id.is_some() {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(link_id.unwrap() as i64)
                .fetch_all(self)
                .await
                .unwrap()
        } else {
            sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .fetch_all(self)
                .await
                .unwrap()
        };
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                row.get::<String, _>("hash"),
                row.get::<i64, _>("link_id") as u64,
            );
        });
        r
    }

    async fn select_uniques_from_index_query(
        &self,
        vk_u: impl UniqueTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        vk_i: VddKey,
    ) -> HashMap<String, u64> {
        let sql = "SELECT hash,link_id FROM t_unique WHERE unique_type = $1 AND link_id IN (SELECT index_item FROM t_index WHERE index_type = $2 AND arg_u64 = $3)";
        let rows = sqlx::query(sql)
            .bind(Into::<u16>::into(vk_u.as_ref()) as i16)
            .bind(Into::<u16>::into(vk_i.as_ref()) as i16)
            .bind(vk_i.get_index_key_as_u64() as i64)
            .fetch_all(self)
            .await
            .unwrap();
        let mut r = HashMap::<String, u64>::new();
        rows.iter().for_each(|row| {
            r.insert(
                row.get::<String, &str>("hash"),
                row.get::<i64, &str>("link_id") as u64,
            );
        });
        r
    }

    async fn save_bound(
        &self,
        vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>,
        bound_content: &str,
    ) -> Ares {
        if !vk.bound_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        sqlx::query(&get_sql("delete_bound", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_bound_left() as i64)
            .bind(vk.get_bound_right_as_u64().unwrap() as i64)
            .execute(self)
            .await
            .unwrap();
        sqlx::query(&get_sql("insert_bound", DataStoreEngine::Postgres))
            .bind(Into::<u16>::into(vk.as_ref()) as i16)
            .bind(vk.get_bound_left() as i64)
            .bind(vk.get_bound_right_as_u64().unwrap() as i64)
            .bind(bound_content)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    /// 多个绑定，根据绑定类型和绑定左值，绑定右值可选
    async fn select_bounds(
        &self,
        vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>,
    ) -> Vec<String> {
        let use_bound_left = vk.get_bound_left() > 0;
        let use_bound_right =
            vk.get_bound_right_as_u64().is_ok() && vk.get_bound_right_as_u64().unwrap() > 0;
        let sql = match (use_bound_left, use_bound_right) {
            (false, false) => {
                "SELECT bound_content FROM t_bound WHERE bound_type = $1"
            }
            (true, false) => {
                "SELECT bound_content FROM t_bound WHERE bound_type = $1 AND bound_left = $2"
            }
            (false, true) => {
                "SELECT bound_content FROM t_bound WHERE bound_type = $1 AND bound_right = $2"
            }
            (true, true) => {
                "SELECT bound_content FROM t_bound WHERE bound_type = $1 AND bound_left = $2 AND bound_right = $3"
            }
        };
        let rows = match (use_bound_left, use_bound_right) {
            (false, false) => sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .fetch_all(self)
                .await
                .unwrap(),
            (true, false) => sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_bound_left() as i64)
                .fetch_all(self)
                .await
                .unwrap(),
            (false, true) => sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_bound_right_as_u64().unwrap() as i64)
                .fetch_all(self)
                .await
                .unwrap(),
            (true, true) => sqlx::query(sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_bound_left() as i64)
                .bind(vk.get_bound_right_as_u64().unwrap() as i64)
                .fetch_all(self)
                .await
                .unwrap(),
        };
        rows.iter()
            .map(|row| row.get::<String, &str>("bound_content"))
            .collect::<Vec<String>>()
    }

    /// 删除多个绑定，根据绑定类型和绑定左值，绑定右值可选
    async fn delete_bounds(&self, vk: impl BoundTypeKeyTrait + Into<u16> + AsRef<VddKey>) -> Ares {
        if !vk.bound_type() {
            return Err(i18n!(E_WRONG_KEY_TYPE));
        }
        let mut sql = "DELETE FROM t_bound WHERE bound_type = $1 AND bound_left = $2".to_string();
        let mut flag_1 = false;
        match vk.get_bound_right_as_u64() {
            Ok(right) => {
                if right > 0 {
                    sql.push_str(" AND bound_right = $3");
                    flag_1 = true
                }
            }
            Err(_) => {}
        }
        let r = if flag_1 {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_bound_left() as i64)
                .bind(vk.get_bound_right_as_u64().unwrap() as i64)
                .execute(self)
                .await
        } else {
            sqlx::query(&sql)
                .bind(Into::<u16>::into(vk.as_ref()) as i16)
                .bind(vk.get_bound_left() as i64)
                .execute(self)
                .await
        };
        r.map_err(|e| e.to_string().into()).map(|_| ())
    }

    async fn count_bounds_grouped(&self, vk_s: &[VddKey]) -> HashMap<u64, usize> {
        if vk_s.len() < 1 {
            return HashMap::new();
        }
        let mut sql_part = String::new();
        vk_s.iter().for_each(|bound_type| {
            sql_part.push_str(&format!(",{}", bound_type.get_bound_left()));
        });
        if vk_s.len() > 0 {
            sql_part.remove(0);
        }
        let sql = format!("SELECT bound_left,count(*) AS c FROM t_bound WHERE bound_type = {} GROUP BY bound_left HAVING bound_left IN ({})", Into::<u16>::into(&vk_s[0]), sql_part);
        let rows = sqlx::query(&sql).fetch_all(self).await.unwrap();
        let mut r = HashMap::<u64, usize>::new();
        rows.iter().for_each(|row| {
            let key = row.get::<i64, _>("bound_left");
            let value = row.get::<i64, _>("c");
            r.insert(key as u64, value as usize);
        });
        r
    }

    async fn u_register(&self, user: &VddUser) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let sql = "INSERT INTO t_entity (entity_id,entity_type,entity_content) VALUES ($1,$2,$3)";
        let vk_e = VddKey::EntityAppUser(user.id.clone());
        let _ = tx
            .execute(
                sqlx::query::<Postgres>(&sql)
                    .bind(user.id.0 as i64)
                    .bind(Into::<u16>::into(&vk_e) as i16)
                    .bind(serde_json::to_string(user).unwrap()),
            )
            .await;
        let sql = "INSERT INTO t_unique (unique_type,hash,link_id) VALUES ($1,$2,$3)";
        let vk_u = VddKey::UniquePhone(user.phone.clone());
        let _ = tx
            .execute(
                sqlx::query::<Postgres>(sql)
                    .bind(Into::<u16>::into(&vk_u) as i16)
                    .bind(user.phone.to_string())
                    .bind(user.id.0 as i64),
            )
            .await;
        tx.commit()
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn m_save_new(&self, model: &VddModel) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("insert_entity", DataStoreEngine::Postgres);
        let vk_e = VddKey::EntityModel(model.id);
        let _ = tx
            .execute(
                sqlx::query(&sql)
                    .bind(model.id as i64)
                    .bind(Into::<u16>::into(vk_e) as i16)
                    .bind(&serde_json::to_string(model).unwrap()),
            )
            .await;
        let sql = get_sql("insert_index", DataStoreEngine::Postgres);
        let vk_i = VddKey::IndexUser2Model(model.uid.clone());
        let _ = tx
            .execute(
                sqlx::query(&sql)
                    .bind(Into::<u16>::into(&vk_i) as i16)
                    .bind(model.id as i64)
                    .bind(vk_i.get_index_key_as_u64() as i64),
            )
            .await;
        let r = tx.commit().await;
        r.map_err(|e| e.to_string().into())
    }

    async fn m_save_tags(
        &self,
        model: &VddModel,
        model_tag: &str,
        tag_map: &HashMap<Astr, u64>,
    ) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let vk_u = VddKey::UniqueModelTag(model_tag.into());
        tx.execute(
            sqlx::query(&get_sql("insert_unique", DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(vk_u) as i16)
                .bind(model_tag)
                .bind(model.id as i64)
                .bind(0i64),
        )
        .await
        .unwrap();
        for x in tag_map.iter() {
            let key = format!("{}:{}", model_tag, x.0);
            let vk_u2 = VddKey::UniqueFieldTag(model_tag.into(), x.0.clone());
            tx.execute(
                sqlx::query(&get_sql("insert_unique", DataStoreEngine::Postgres))
                    .bind(Into::<u16>::into(&vk_u2) as i16)
                    .bind(key)
                    .bind(*x.1 as i64)
                    .bind(model.id as i64),
            )
            .await
            .unwrap();
        }
        let r = tx.commit().await;
        r.map_err(|e| e.to_string().into())
    }

    async fn f_save_new(&self, field: &VddField) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("insert_entity", DataStoreEngine::Postgres);
        let vk_e = VddKey::EntityField(field.id.clone());
        tx.execute(
            sqlx::query(&sql)
                .bind(field.id.0 as i64)
                .bind(Into::<u16>::into(&vk_e) as i16)
                .bind(serde_json::to_string(field).unwrap()),
        )
        .await
        .unwrap();
        let sql = get_sql("insert_index", DataStoreEngine::Postgres);
        let vk_i = VddKey::IndexModel2Field(field.model_id);
        tx.execute(
            sqlx::query(&sql)
                .bind(Into::<u16>::into(&vk_i) as i16)
                .bind(field.id.0 as i64)
                .bind(vk_i.get_index_key_as_u64() as i64),
        )
        .await
        .unwrap();
        let r = tx.commit().await;
        r.map_err(|e| e.to_string().into())
    }

    async fn f_delete(&self, model_id: u64, field_id: FieldId) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let sql = get_sql("delete_index", DataStoreEngine::Postgres);
        let vk_i = VddKey::IndexModel2Field(model_id);
        tx.execute(
            sqlx::query(&sql)
                .bind(Into::<u16>::into(&vk_i) as i16)
                .bind(field_id.0 as i64)
                .bind(vk_i.get_index_key_as_u64() as i64),
        )
        .await
        .map_err(|e| Arc::from(e.to_string()))
        .map(|_| ())?;
        let sql = get_sql("delete_entity", DataStoreEngine::Postgres);
        tx.execute(sqlx::query(&sql).bind(field_id.0 as i64))
            .await
            .map_err(|e| Arc::from(e.to_string()))
            .map(|_| ())?;
        tx.commit()
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn m_clear_tags(&self, model_id: u64, field_ids: &[FieldId]) -> Ares {
        let mut tx = self.begin().await.unwrap();
        let vk_u = VddKey::UniqueModelTag("".into());
        let vk_u2 = VddKey::UniqueFieldTag("".into(), "".into());
        tx.execute(
            sqlx::query(&get_sql("delete_unique", DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(&vk_u) as i16)
                .bind(model_id as i64),
        )
        .await
        .unwrap();
        for x in field_ids {
            tx.execute(
                sqlx::query(&get_sql("delete_unique", DataStoreEngine::Postgres))
                    .bind(Into::<u16>::into(&vk_u2) as i16)
                    .bind(x.0 as i64),
            )
            .await
            .unwrap();
        }
        tx.commit().await.map_err(|e| e.to_string().into())
    }

    async fn do_delete_entity(&self, id: u64) -> Result<(), Astr> {
        sqlx::query(&get_sql("do_delete_entity", DataStoreEngine::Postgres))
            .bind(id as i64)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn restore_entity(&self, id: u64) -> Ares {
        sqlx::query(&get_sql("restore_entity", DataStoreEngine::Postgres))
            .bind(id as i64)
            .execute(self)
            .await
            .map_err(|e| e.to_string().into())
            .map(|_| ())
    }

    async fn check_entity_deleted(&self, id: u64) -> bool {
        let sql_r = sqlx::query(&get_sql("check_entity_deleted", DataStoreEngine::Postgres))
            .bind(id as i64)
            .fetch_one(self)
            .await
            .unwrap();
        sql_r.get::<i64, &str>("count") as usize > 0
    }
    async fn save_plan(&self, plan: &QueryPlan, is_update: bool) -> Ares {
        let hash = to_sha256_hash(plan.name.as_bytes());
        let vk_u = VddKey::UniqueQueryPlan(plan.u_id.clone(), hash.clone());
        self.delete_unique_by_value(vk_u.clone(), plan.id).await?;
        let vk_b = VddKey::BoundQueryPlan(plan.u_id.clone(), plan.id);
        self.delete_bounds(vk_b.clone()).await?;
        let id2 = self.get_unique(vk_u.clone()).await;
        if id2.is_ok() && id2.unwrap() == plan.id && !is_update {
            return Err("已有同名称查询计划存在".into());
        }
        let mut tx = self.begin().await.unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_bound", DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(&vk_b) as i16)
                .bind(vk_b.get_bound_left() as i64)
                .bind(vk_b.get_bound_right_as_u64().unwrap() as i64)
                .bind(json!(plan).to_string()),
        )
        .await
        .unwrap();
        tx.execute(
            sqlx::query(&get_sql("insert_unique", DataStoreEngine::Postgres))
                .bind(Into::<u16>::into(&vk_u) as i16)
                .bind(hash.to_string())
                .bind(plan.id as i64)
                .bind(plan.u_id.0 as i64),
        )
        .await
        .unwrap();
        tx.commit().await.map_err(|e| e.to_string().into())
    }
}

pub(crate) async fn test_connection(ds: &DataStore) -> bool {
    get_pg_pool(ds)
        .await
        .acquire()
        .await
        .unwrap()
        .ping()
        .await
        .is_ok()
}

pub(crate) async fn table_to_fields(model: &VddModel) -> Vec<VddFieldRef> {
    let store = model.store.clone().unwrap();
    let pool = get_pg_pool(store.as_ref()).await;
    let table_name = store.remote_key.split(".").collect::<Vec<&str>>();
    let p = if table_name.len() == 1 {
        (table_name[0], "public")
    } else {
        (table_name[1], table_name[0])
    };
    let rows = sqlx::query(
        r#"
SELECT
column_name,
data_type,
character_maximum_length,
is_nullable,
column_default
FROM
information_schema.columns
WHERE
table_name = $1 AND table_schema = $2;"#,
    )
    .bind(p.0.to_string())
    .bind(p.1.to_string())
    .fetch_all(&pool)
    .await
    .unwrap();
    let mut field_vec = Vec::<VddFieldRef>::new();
    let mut display_order = 1;
    for row in rows.iter() {
        let field_type = row.try_get::<&str, _>("data_type").unwrap();
        let col_name: &str = row.try_get("column_name").unwrap();
        let field = VddField::new(
            model.id,
            display_order,
            col_name,
            Into::<u16>::into(&type_support::judge_field_type(field_type)),
        );
        field_vec.push(Arc::new(field));
        display_order = display_order + 1;
    }
    field_vec
}

fn gen_sql_for_page(schema: &VddSchema, q: &WebQueryParams) -> String {
    let mut cols = String::new();
    schema.fields.iter().for_each(|field| {
        let col_name = schema.get_field_tag(&field.id);
        cols.push_str(&format!(",{}", col_name));
    });
    cols.remove(0);
    let mut filters = String::new();
    q.filters.iter().for_each(|x| match x.as_ref() {
        QueryFilter::QfFiledValue(fve) => {
            let field_id = fve.get_field_id();
            let field = schema.try_get_field(&field_id).unwrap();
            let part =
                fve.to_postgsql_where_part(&schema.get_field_tag(&field_id), field.quote_value());
            if !part.is_empty() {
                filters.push_str(" ");
                filters.push_str(&part);
            }
        }
        _ => {}
    });
    // 只获取逻辑有效的记录值
    match schema.get_model_logic_useful_sql() {
        Some(lu_sql) => {
            filters.push_str(if filters.is_empty() { " " } else { " AND " });
            filters.push_str(&lu_sql);
        }
        None => {
            filters.push_str(if filters.is_empty() {
                " 1 = 1"
            } else {
                " AND 1 = 1"
            });
        }
    }
    let table_name = schema.get_model_remote_key().unwrap();
    match schema.get_model_primary_key() {
        Some(pk_col) => {
            format!(
                "SELECT {0} FROM {1} WHERE {2} > {3} AND {5} ORDER BY {2} ASC LIMIT {4}",
                cols, table_name, pk_col, q.last_id, q.page_size, filters
            )
        }
        None => {
            format!(
                "SELECT {0} FROM {1} WHERE {3} LIMIT {2}",
                cols, table_name, q.page_size, filters
            )
        }
    }
}

async fn rows_to_page(
    pool: &PgPool,
    data_page_sql: &str,
    s: &VddSchema,
    q: &WebQueryParams,
) -> QueryDataPage<VddDisplay> {
    let mut records = Vec::<Arc<VddDisplay>>::new();
    let mut last_index = q.last_id;
    let mut rows = sqlx::query::<Postgres>(data_page_sql).fetch(pool);
    while let Some(row) = rows.try_next().await.unwrap() {
        let mut record = VddRecord::new_default(s.model.uid.clone(), s.model.id);
        type_support::row_to_record(&row, &mut record, &s.fields, s.get_model_primary_key());
        last_index = record.meta.id.0;
        let recordd = record.new_display(s).await;
        records.push(Arc::new(recordd));
        if records.len() >= q.page_size as usize {
            break;
        }
    }
    QueryDataPage {
        data: records,
        all_record_count: 0,
        last_index,
    }
}

pub(crate) async fn get_page_sqlx(
    schema: &VddSchema,
    q: &WebQueryParams,
) -> QueryDataPage<VddDisplay> {
    let store = schema.model.store.clone().unwrap();
    let pool = get_pg_pool(store.as_ref()).await;
    let d_sql = gen_sql_for_page(schema, q);
    let mut r = rows_to_page(&pool, &d_sql, schema, q).await;
    r.all_record_count = 0;
    r
}

pub(crate) async fn share_page(sqp: &ShareQueryParams) -> QueryDataPage<VddDisplay> {
    let store = sqp.schema.model.store.clone().unwrap();
    let pool = get_pg_pool(store.as_ref()).await;
    let sql = gen_sql_for_page(&sqp.schema, &sqp.params);
    rows_to_page(&pool, &sql, &sqp.schema, &sqp.params).await
}

mod type_support {
    use std::sync::Arc;

    use rust_decimal::prelude::ToPrimitive;
    use rust_decimal::Decimal;
    use sqlx::postgres::types::PgTimeTz;
    use sqlx::postgres::PgRow;
    use sqlx::Column;
    use sqlx::Row;
    use visit_dd_core::value::VddValue;
    use visit_dd_core::RecordId;
    use visit_dd_core::DT_FORMAT;
    use visit_dd_core::T_FORMAT;
    use visit_dd_core::{
        field::{VddField, VddFieldType},
        record::VddRecord,
        Astr,
    };

    pub(super) fn judge_field_type(s: &str) -> VddFieldType {
        tracing::info!("judge field type:{}", s);
        if s.starts_with("int") {
            VddFieldType::FtInt {
                min_v: None,
                max_v: None,
                unit_label: None,
            }
        } else if s.eq_ignore_ascii_case("timestamp") || s.eq_ignore_ascii_case("datetime") {
            VddFieldType::FtDateTime("".into())
        } else {
            VddFieldType::FtString {
                min_len: None,
                max_len: None,
                pattern: None,
            }
        }
    }

    pub(super) fn row_to_record(
        row: &PgRow,
        record: &mut VddRecord,
        fields: &[Arc<VddField>],
        pk: Option<Astr>,
    ) {
        row.columns().iter().for_each(|col| {
            let col_name = col.name();
            let col_type = col.type_info().to_string();
            let v = match col_type.as_str() {
                "INT8" => match row.try_get::<i64, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S201(t)),
                },
                "INT4" => match row.try_get::<i32, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S201(t as i64)),
                },
                "BOOL" => match row.try_get::<bool, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S211(t)),
                },
                "FLOAT8" => match row.try_get::<f64, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S202(t)),
                },
                "TIMESTAMPTZ" => match row.try_get::<chrono::NaiveDateTime, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S101(t.format(DT_FORMAT).to_string())),
                },
                "TIMETZ" => match row.try_get::<PgTimeTz, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S101(t.time.format(T_FORMAT).to_string())),
                },
                "NUMERIC" => match row.try_get::<Decimal, _>(col_name) {
                    Err(e) => Arc::new(VddValue::S101(e.to_string())),
                    Ok(t) => Arc::new(VddValue::S202(t.to_f64().unwrap())),
                },
                _ => {
                    println!("Col Type Pg {:?}", col_type);
                    match row.try_get::<&str, _>(col_name) {
                        Err(e) => Arc::new(VddValue::S101(e.to_string())),
                        Ok(t) => Arc::new(VddValue::S101(t.to_string())),
                    }
                }
            };
            let field = fields
                .iter()
                .filter(|f| f.label == col_name)
                .next()
                .unwrap();
            record.value.insert(field.id.clone(), v.clone());
            //
            if let Some(pk2) = pk.clone() {
                if !pk2.is_empty() && field.label == pk2.as_ref() {
                    record.set_id(&TryInto::<RecordId>::try_into(v.as_ref()).unwrap());
                }
            }
        });
    }
}
